package com.zengjianjun.es.demo.config;

import com.zengjianjun.es.demo.highlevelclient.interceptor.ElasticsearchRequestInterceptor;
import com.zengjianjun.es.demo.highlevelclient.interceptor.ElasticsearchResponseInterceptor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.client.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.elasticsearch.client.ClientConfiguration;
import org.springframework.data.elasticsearch.client.RestClients;

/**
 * es高阶客户端配置
 *
 * @author zengjianjun
 */
@Slf4j
@Configuration
public class EsConfig {

    @Value("${spring.elasticsearch.uris}")
    private String uris;

    @Value("${spring.elasticsearch.username}")
    private String username;

    @Value("${spring.elasticsearch.password}")
    private String password;

    public static final RequestOptions COMMON_OPTIONS;

    static {
        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
        // 默认缓存限制为100MB，此处修改为30MB。
        builder.setHttpAsyncResponseConsumerFactory(
                new HttpAsyncResponseConsumerFactory
                        .HeapBufferedResponseConsumerFactory(30 * 1024 * 1024));
        COMMON_OPTIONS = builder.build();
    }

    /**
     * 加上下方日志配置，可以用于打印es执行的dsl语句与响应结果
     * # es日志配置
     * logging.level.org.springframework.data.elasticsearch.client.WIRE: trace
     */
    @Bean(destroyMethod = "close")
    public RestHighLevelClient restClient() {
        String hostAndPost = uris;
        if (uris.contains("http://")) {
            hostAndPost = uris.replace("http://", "");
        } else if (uris.contains("https://")) {
            hostAndPost = uris.replace("https://", "");
        }
        ClientConfiguration clientConfiguration = ClientConfiguration.builder()
                .connectedTo(hostAndPost)
                .build();
        return RestClients.create(clientConfiguration).rest();
    }

    /**
     * 用于阿里云es集群高阶客户端构建
     */
//    @Bean
    public RestHighLevelClient restHighLevelClient() {
        log.info("uris= {} , username={} , password={}", uris, username, password);
        String[] uriArray = StringUtils.split(uris, ",");
        String[] hostArray = new String[uriArray.length];
        for (int i = 0; i < uriArray.length; i++) {
            String uri = uriArray[i];
            int startIndex = uri.indexOf("//");
            int endIndex = uri.lastIndexOf(":");
            if (endIndex < 0) {
                endIndex = uri.length();
            }
            hostArray[i] = uri.substring(startIndex + 2, endIndex);
        }
        int port = Integer.parseInt(uriArray[0].substring(uriArray[0].lastIndexOf(":") + 1));
        // 阿里云Elasticsearch集群需要basic auth验证。
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        // 访问用户名和密码为您创建阿里云Elasticsearch实例时设置的用户名和密码，也是Kibana控制台的登录用户名和密码。
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
        // 通过builder创建rest client，配置http client的HttpClientConfigCallback。
        // 单击所创建的Elasticsearch实例ID，在基本信息页面获取公网地址，即为ES集群地址。
        RestClientBuilder builder = RestClient.builder(new HttpHost(StringUtils.join(hostArray, ","), port, "http"))
                .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                        httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                        httpClientBuilder.setKeepAliveStrategy((response, context) -> 1000 * 60);
                        return httpClientBuilder;
                    }
                }).setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
                    // 该方法接收一个RequestConfig.Builder对象，对该对象进行修改后然后返回。
                    @Override
                    public RequestConfig.Builder customizeRequestConfig(
                            RequestConfig.Builder requestConfigBuilder) {
                        return requestConfigBuilder
                                .setConnectTimeout(5000 * 1000) // 连接超时（默认为1秒）
                                .setSocketTimeout(6000 * 1000);// 套接字超时（默认为30秒） // 更改客户端的超时限制默认30秒现在改为100*1000分钟
                    }
//                }).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
//                    @Override
//                    public HttpAsyncClientBuilder customizeHttpClient(
//                            HttpAsyncClientBuilder httpClientBuilder) {
//                        return httpClientBuilder.setDefaultIOReactorConfig(
//                                IOReactorConfig.custom()
//                                        .setIoThreadCount(10) // 设置线程数
//                                        .build());
//                    }
//                });
                });
        // 添加自定义拦截器
        builder.setHttpClientConfigCallback(httpClientBuilder -> {
            httpClientBuilder.addInterceptorFirst(new ElasticsearchRequestInterceptor());
            httpClientBuilder.addInterceptorFirst(new ElasticsearchResponseInterceptor());
            return httpClientBuilder;
        });
        String auth = Base64.encodeBase64String((username + ":" + password).getBytes());
        builder.setDefaultHeaders(new BasicHeader[]{new BasicHeader("Authorization", "Basic " + auth)});
        // RestHighLevelClient实例通过REST low-level client builder进行构造。
        return new RestHighLevelClient(builder);
    }
}
